package com.atguigu.day4

import com.atguigu.source.{SensorReading, SensorSource}
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object FreezingAlarm {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)
    //

    val stream = env.addSource(new SensorSource)
      //没有开窗
      .process(new FreezingAlarmFunction)

//    stream.print()
    stream.getSideOutput(new OutputTag[String]("freesing-alar")).print()
    env.execute()
  }

  class FreezingAlarmFunction extends ProcessFunction[SensorReading,SensorReading]{
    //定义一个侧输出标签，实际上就是侧输出流的名字
    lazy val freezingAlarmout = new OutputTag[String]("freesing-alar")

    override def processElement(value: SensorReading, ctx: ProcessFunction[SensorReading, SensorReading]#Context, out: Collector[SensorReading]): Unit = {
        if(value.temperature<32){
          ctx.output(freezingAlarmout,s"${value.id}     低温警告")
        }
      //常规温度正常输出
      out.collect(value)

    }
  }
}
